package ex.datastream.functions.function;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;

public class MyAggregateFunction  implements AggregateFunction<Tuple2<String, Long>, Tuple2<String, Long>, Tuple2<String, Long>> {
    //初始化
    @Override
    public Tuple2<String, Long> createAccumulator() {
        return new Tuple2<>("", 0L);
    }

    //按key,累加value
    @Override
    public Tuple2<String, Long> add(Tuple2<String, Long> stringLongTuple2, Tuple2<String, Long> stringLongTuple22) {
        return new Tuple2<>(stringLongTuple2.f0 , stringLongTuple2.f1 + stringLongTuple22.f1);
    }

    //返回结果
    @Override
    public Tuple2<String, Long> getResult(Tuple2<String, Long> stringLongTuple2) {
        return stringLongTuple2;
    }

    @Override
    public Tuple2<String, Long> merge(Tuple2<String, Long> stringLongTuple2, Tuple2<String, Long> acc1) {
        return new Tuple2<>(stringLongTuple2.f0 , stringLongTuple2.f1);
    }
}